package tul.cot.job.output;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang.NotImplementedException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

/**
 * This class enables to read all files from a given directory (local fs, hdfs).
 */
public abstract class OutputReader {
	
	protected final Configuration conf;
	protected final File outDir;
	
	protected OutputReader(Configuration conf, File outDir) {
		this.conf = conf;
		this.outDir = outDir;
	}
	
	/**
	 * Create {@code OutputReader} for the specified directory.
	 * Boolean flag determines whether it reads from local fs or hdfs.
	 * @throws IOException 
	 */
	public static OutputReader create(Configuration conf, File outDir) throws IOException {
		return new FSOutputReader(conf, outDir);
	}
	
	/**
	 * @return the iterable object over lines from all files
	 */
	public abstract Iterable<String> getLines();
	
}

/**
 * Implementation of the {@code OutputReader} for file system from {@link Configuration}.
 */
class FSOutputReader extends OutputReader {
	
	public FSOutputReader(Configuration conf, File outDir) throws IOException {
		super(conf, outDir);
		init();
	}
	
	private FileSystem fs;
	private Iterator<Path> outFiles;
	private BufferedReader reader;
	private String nextLine;
	
	
	private final Iterator<String> linesIter = this.new LinesIterator();
	private final Iterable<String> lines = new Iterable<String>() {
		
		@Override
		public Iterator<String> iterator() {
			return linesIter;
		}
	};

	private class LinesIterator implements Iterator<String> {

		@Override
		public boolean hasNext() {
			return nextLine != null;
		}

		@Override
		public String next() {
			String line = nextLine;
			nextLine = readNext();
			return line;
		}

		@Override
		public void remove() {
			throw new NotImplementedException();
		}
		
	}
	
	private void init() throws IOException {
		fs = FileSystem.get(conf);
		
		PathFilter filter = new PathFilter() {
			
			@Override
			public boolean accept(Path path) {
				return path.getName().startsWith("part");
			}
		};
		List<Path> paths = Arrays.asList(FileUtil.stat2Paths(fs.listStatus(new Path(outDir.toString()), filter)));
		
		
		outFiles = paths.iterator();
		
		if (outFiles.hasNext()) {
			try {
				reader = new BufferedReader(new InputStreamReader(fs.open(outFiles.next())));
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		} else {
			reader = null;
		}
		
		nextLine = readNext();
	}
	
	private String readNext() {
		
		try {
			if (reader == null)
				return null;
			
			String ret = reader.readLine();
			if (ret != null)
				return ret;
			
			while (outFiles.hasNext()) {
				reader = new BufferedReader(new InputStreamReader(fs.open(outFiles.next())));
				ret = reader.readLine();
				if (ret != null)
					return ret;
			}
			
			return null;
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
	
	@Override
	public Iterable<String> getLines() {
		return lines;
	}
	
}


/**
 * Implementation of the {@code OutputReader} for local file system.
 */
class LocalOutputReader extends OutputReader {
	
	public LocalOutputReader(File outDir) {
		super(null, outDir);
		init();
	}
	
	private Iterator<File> outFiles;
	private BufferedReader reader;
	private String nextLine;
	
	
	private final Iterator<String> linesIter = this.new LinesIterator();
	private final Iterable<String> lines = new Iterable<String>() {
		
		@Override
		public Iterator<String> iterator() {
			return linesIter;
		}
	};

	private class LinesIterator implements Iterator<String> {

		@Override
		public boolean hasNext() {
			return nextLine != null;
		}

		@Override
		public String next() {
			String line = nextLine;
			nextLine = readNext();
			return line;
		}

		@Override
		public void remove() {
			throw new NotImplementedException();
		}
		
	}
	
	private void init() {
		List<File> files = new ArrayList<File>();
		for (File f: outDir.listFiles()) {
			if (f.isFile())
				files.add(f);
		}
		
		outFiles = files.iterator();
		
		if (outFiles.hasNext()) {
			try {
				reader = new BufferedReader(new FileReader(outFiles.next()));
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		} else {
			reader = null;
		}
		
		nextLine = readNext();
	}
	
	private String readNext() {
		
		try {
			if (reader == null)
				return null;
			
			String ret = reader.readLine();
			if (ret != null)
				return ret;
			
			while (outFiles.hasNext()) {
				reader = new BufferedReader(new FileReader(outFiles.next()));
				ret = reader.readLine();
				if (ret != null)
					return ret;
			}
			
			return null;
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
	
	@Override
	public Iterable<String> getLines() {
		return lines;
	}
	
}